package com.lcifn.middleware.rpc.transport;

import com.lcifn.middleware.rpc.transport.command.Command;
import com.lcifn.middleware.rpc.transport.future.InFlightRequest;
import com.lcifn.middleware.rpc.transport.future.ResponseFuture;
import io.netty.channel.Channel;

import java.util.concurrent.CompletableFuture;

/**
 * 传输服务netty实现
 * @author lichao
 * @date 2021/1/6 17:08
 */
public class NettyTransport implements Transport {

    private Channel channel;
    private InFlightRequest inFlightRequest;

    NettyTransport(Channel channel, InFlightRequest inFlightRequest) {
        this.channel = channel;
        this.inFlightRequest = inFlightRequest;
    }

    /**
     * 发送请求指令
     *
     * @param request 指令请求
     * @return
     */
    @Override
    public CompletableFuture<Command> send(Command request) {
        int requestId = request.getHeader().getRequestId();
        // 构建返回值
        CompletableFuture<Command> completableFuture = new CompletableFuture<>();
        inFlightRequest.add(new ResponseFuture(requestId, completableFuture));
        try {
            channel.writeAndFlush(request).addListener(channelFuture -> {
                if (!channelFuture.isSuccess()) {
                    inFlightRequest.remove(requestId);
                    completableFuture.completeExceptionally(channelFuture.cause());
                    channel.close();
                }
            });
        } catch (Throwable t) {
            inFlightRequest.remove(requestId);
            completableFuture.completeExceptionally(t);
        }
        return completableFuture;
    }
}
